/*
 * 文件名：PassivDataSendThread.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年11月14日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.model.RealDataDispatcher;

/**
 * @description 数据发送线程 各个数据单元对应一个发送线程
 * @author [yuanze]
 * @version [版本号,2014年11月14日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class DataSendThread extends Thread
{
	private static Logger logger = LoggerFactory.getLogger(DataSendThread.class);

	private RealDataDispatcher dispatcher;

	public DataSendThread(RealDataDispatcher dispatcher)
	{
		this.dispatcher = dispatcher;
	}

	/**
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run()
	{
		logger.info("单元接口[" + dispatcher.getRealStreamUnit().getOnlyUnitCode() + "]发送消息线程启动成功.........");
		dispatcher.setRunning(true);
		logger.info("单元接口[" + dispatcher.getRealStreamUnit().getOnlyUnitCode() + "]分发线程启动");
		while (dispatcher.isSendable())
		{
			// 被动模式(不保证所有的channel都能接受到数据)
			if (dispatcher.isPassiveSend())
			{
				passiveModeProcess();
			}
			else
			{
				initiativeModeProcess();
			}
		}
		dispatcher.setRunning(false);
	}

	/**
	 * 
	 * @description 被动模式
	 */
	private void passiveModeProcess()
	{
		// try
		// {
		// RealStreamDataPack dpack = dispatcher.takeMessage();
		// boolean sendMsgIsNormal = false;
		// for (Map.Entry<String, Channel> channelKey :
		// dispatcher.getSEND_CHANNEL_CACHE().entrySet())
		// {
		// Channel channel = channelKey.getValue();
		// // 提供发送效率减少判断
		// if (!channel.isActive())// || !channel.isWritable())
		// {
		// dispatcher.getSEND_CHANNEL_CACHE().remove(channelKey.getKey());
		// }
		// else
		// {
		// channel.writeAndFlush(dpack);
		// sendMsgIsNormal = true;
		// logger.debug("单元接口[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]发送消息成功.........");
		// }
		// }
		// if (!sendMsgIsNormal)
		// {
		// dispatcher.addMessage(dpack);
		// dispatcher.setSendable(false);
		// logger.warn("单元接口[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]监测到数据没有可以发送的通道 停止发送");
		// }
		// }
		// catch (InterruptedException e)
		// {
		// logger.error("被动模式(不保证所有的channel都能接受到数据)发送异常", e);
		// }
	}

	/**
	 * 
	 * @description 主动模式处理
	 */
	private void initiativeModeProcess()
	{
		// try
		// {
		// RealStreamDataPack dpack = dispatcher.takeMessage();
		// // 获得主动推送远程地址 首次进入建立channel连接
		// List<SocketInfo> socketInfoList =
		// dispatcher.getRealStreamUnit().getSocketInfo();
		// for (SocketInfo socketInfo : socketInfoList)
		// {
		// String key = socketInfo.getSocketIp() + socketInfo.getSocketPort();
		// if (!dispatcher.getSEND_CHANNEL_CACHE().contains(key))
		// {
		// Channel channel = new
		// StreamDataInitiativeSendClient(socketInfo.getSocketIp(),
		// socketInfo.getSocketPort()).run();
		// dispatcher.getSEND_CHANNEL_CACHE().put(key, channel);
		// }
		// }
		//
		// boolean sendMsgIsNormal = true;
		// for (Map.Entry<String, Channel> channelKey :
		// dispatcher.getSEND_CHANNEL_CACHE().entrySet())
		// {
		// Channel channel = channelKey.getValue();
		// if (channel == null || !channel.isActive() || !channel.isWritable())
		// {
		// dispatcher.getSEND_CHANNEL_CACHE().remove(channelKey.getKey());
		// sendMsgIsNormal = false;
		// }
		// else
		// {
		// channel.writeAndFlush(dpack);
		// }
		// }
		// // 主动模式监测到不是所有的channel发送成功
		// if (!sendMsgIsNormal)
		// {
		// dispatcher.addMessage(dpack);
		// dispatcher.setSendable(false);
		// logger.warn("单元接口[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]监测到数据没有可以发送的通道 停止发送");
		// }
		// }
		// catch (InterruptedException e)
		// {
		// logger.error("被动模式(不保证所有的channel都能接受到数据)发送异常", e);
		// }
	}
}
